package com.bill99.rpc.registry.zookeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.I0Itec.zkclient.exception.ZkNoNodeException;

import com.bill99.rpc.RpcException;
import com.bill99.rpc.common.Constants;
import com.bill99.rpc.common.util.UrlUtils;
import com.bill99.rpc.registry.BaseUrl;
import com.bill99.rpc.registry.NotifyListener;
import com.bill99.rpc.registry.support.AbstractRegistry;
import com.bill99.rpc.registry.url.ReferenceUrl;
import com.bill99.rpc.registry.url.RegistryUrl;
import com.bill99.rpc.registry.url.ServiceUrl;
import com.bill99.rpc.registry.zookeeper.client.ZookeeperClient;

/**
 * 
 * zookeeper存储结构 
 * 根节点 			/bill99 
 * Service节点		 /com.bill99.DemoService 
 * 服务提供者 				/providers
 * 服务提供者URL 				/hessian://192.168.1.1:8080/rpc-demo?application=provider-demo&interface=com.bill99.DemoService 
 * 服务消费者 				/consumers 
 * 服务消费者URL					/hessian://192.168.51.1/application=consumer-demo&interface=com.bill99.DemoService
 * 
 * 流程说明： 服务提供者启动时 向/bill99/com.bill99.DemoService/providers目录下写入自己的URL地址。
 * 服务消费者启动时 订阅/bill99/com.bill99.DemoService/providers目录下的提供者URL地址。
 * 并向/bill99/com.bill99.DemoService/consumers目录下写入自己的URL地址。
 * 
 * @author sheng.zhang
 * /bill99-rpc
 */
public class ZookeeperRegistry extends AbstractRegistry {

	private final ConcurrentMap<ReferenceUrl, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<ReferenceUrl, ConcurrentMap<NotifyListener, ChildListener>>();

	private final ZookeeperClient zkClient;

	public ZookeeperRegistry(RegistryUrl registryUrl,
			ZookeeperTransporter zookeeperTransporter) {
		super(registryUrl);
		zkClient = zookeeperTransporter.connect(registryUrl);
		zkClient.addStateListener(new StateListener() {
			public void stateChanged(int state) {
				if (state == RECONNECTED) {
					try {
						// recover();
					} catch (Exception e) {
						logger.error(e.getMessage(), e);
					}
				}
			}
		});
	}

	@Override
	public boolean isAvailable() {
		return zkClient.isConnected();
	}

	@Override
	public void destroy() {
		try {
			if (zkClient != null)
				zkClient.close();
		} catch (Exception e) {
			logger.warn("Failed to close zookeeper client " + getRegistryUrl()
					+ ", cause: " + e.getMessage(), e);
		}
	}

	@Override
	public RegistryUrl getRegistryUrl() {
		return registryUrl;
	}

	protected void doRegister(ServiceUrl serviceUrl) {
		try {
			zkClient.create(toUrlPath(serviceUrl), false);
		} catch (Throwable e) {
			throw new RpcException("Failed to register " + serviceUrl
					+ " to zookeeper " + getRegistryUrl() + ", cause: "
					+ e.getMessage(), e);
		}
	}

	protected void doUnregister(ServiceUrl serviceUrl) {
		try {
			zkClient.delete(toUrlPath(serviceUrl));
		} catch (Throwable e) {
			throw new RpcException("Failed to unregister " + serviceUrl
					+ " to zookeeper " + getRegistryUrl() + ", cause: "
					+ e.getMessage(), e);
		}
	}
	
	protected void doSubscribe(final ReferenceUrl referenceUrl,
			final NotifyListener listener) {
		try {
			ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners
					.get(referenceUrl);
			if (listeners == null) {
				listeners = new ConcurrentHashMap<NotifyListener, ChildListener>();
				zkListeners
						.putIfAbsent(
								referenceUrl,
								listeners);
			}
			ChildListener zkListener = listeners.get(listener);
			if (zkListener == null) {
				zkListener = new ChildListener() {
					@Override
					public void childChanged(String path, List<String> children) {
						logger.info("zookeeper child has changed,path: {}",path);
						List<ServiceUrl> serviceUrls = lookup(referenceUrl);
						listener.notify(serviceUrls);
					}
				};
				listeners.putIfAbsent(listener, zkListener);
			}
			// 创建消费端
			zkClient.create(toUrlPath(referenceUrl), false);
			// 对consumer对应的providers目录进行监听
			zkClient.addChildListener(toProviderPath(referenceUrl), zkListener);
		} catch (Throwable e) {
			throw new RpcException("Failed to subscribe " + referenceUrl
					+ " to zookeeper " + getRegistryUrl() + ", cause: "
					+ e.getMessage(), e);
		}
	}

	protected void doUnsubscribe(ReferenceUrl referenceUrl, NotifyListener listener) {
		zkClient.delete(toUrlPath(referenceUrl));
		ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners
				.get(referenceUrl);
		if (listeners != null) {
			ChildListener zkListener = listeners.get(listener);
			if (zkListener != null) {
				zkClient.removeChildListener(toProviderPath(referenceUrl), zkListener);
			}
		}
	}

	@Override
	protected List<ServiceUrl> doLookup(ReferenceUrl referenceUrl) {
		if (referenceUrl == null) {
			throw new IllegalArgumentException("lookup referenceUrl == null");
		}
		try {
			List<String> providers = new ArrayList<String>();
			try {
				String urlPath = toProviderPath(referenceUrl);
				List<String> children = zkClient.getChildren(urlPath);
				if (children != null) {
					providers.addAll(children);
				}
			} catch (ZkNoNodeException e) {
				// ignore
			}
			return toServiceUrls(referenceUrl, providers);
		} catch (Throwable e) {
			throw new RpcException("Failed to lookup " + referenceUrl
					+ " from zookeeper " + getRegistryUrl() + ", cause: "
					+ e.getMessage(), e);
		}
	}
	
	//~~~~~~~~~~ url path ~~~~~~~~~~~~~~//
	
	/**
	 * @return "/bill99-rpc/"
	 */
	private String toRootDir() {
		return Constants.ROOT_ZOOKEEPER + Constants.PATH_SEPARATOR;
	}
	
	/**
	 * @param url
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService
	 */
	private String toServicePath(BaseUrl url) {
		String name = url.getServiceInterface();
		return toRootDir() + UrlUtils.encode(name);
	}
	
	/**
	 * @param serviceUrl
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService/providers"
	 */
	private String toProviderPath(ServiceUrl serviceUrl) {
		return toServicePath(serviceUrl)
				+ Constants.PATH_SEPARATOR
				+ Constants.PROVIDERS_CATEGORY;
	}
	
	/**
	 * @param referenceUrl
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService/providers"
	 */
	private String toProviderPath(ReferenceUrl referenceUrl) {
		return toServicePath(referenceUrl)+Constants.PATH_SEPARATOR+Constants.PROVIDERS_CATEGORY;
	}

	/**
	 * @param serviceUrl
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService/consumers"
	 */
	private String toConsumerPath(ReferenceUrl referenceUrl) {
		return toServicePath(referenceUrl)
				+ Constants.PATH_SEPARATOR
				+ Constants.CONSUMERS_CATEGORY;
	}
	
	/**
	 * @param serviceUrl
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService/providers/hessian://127.0.0.1:8080/rpc-demo/demoService?interface=com.bill99.rpc.demo.DemoService&..."
	 */
	private String toUrlPath(ServiceUrl serviceUrl) {
		return toProviderPath(serviceUrl) + Constants.PATH_SEPARATOR
				+ UrlUtils.encode(serviceUrl.toFullString());
	}
	
	/**
	 * @param referenceUrl
	 * @return "/bill99-rpc/com.bill99.rpc.demo.DemoService/consumers/hessian://127.0.0.1/interface=com.bill99.rpc.demo.DemoService&..."
	 */
	private String toUrlPath(ReferenceUrl referenceUrl) {
		return toConsumerPath(referenceUrl) + Constants.PATH_SEPARATOR
				+ UrlUtils.encode(referenceUrl.toFullString());
	}

	/**
	 * 将providers路径下的字符串转化为ServiceUrl对象
	 * @param referenceUrl
	 * @param providers
	 * @return
	 */
	private List<ServiceUrl> toServiceUrls(ReferenceUrl referenceUrl,
			List<String> providers) {
		List<ServiceUrl> urls = new ArrayList<ServiceUrl>();
		if (providers != null && providers.size() > 0) {
			for (String provider : providers) {
				provider = UrlUtils.decode(provider);
				if (provider.contains("://")) {
					ServiceUrl serviceUrl = ServiceUrl.valueOf(provider);
					urls.add(serviceUrl);
				}
			}
		}
		return urls;
	}

}